package com.navinfo.platform.score.service.impl;

import com.fasterxml.jackson.databind.JsonMappingException;
import com.navinfo.platform.score.pojo.ScoreWeight;
import com.navinfo.platform.score.service.LoadScoreWeightService;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.io.IOException;
import java.io.Serializable;
import java.math.BigDecimal;
import java.sql.*;
import java.util.*;

/**
 * 加载
 * @author CAI
 */
public class LoadScoreWeightDBServiceImpl implements LoadScoreWeightService, Serializable {
    static final Logger logger = LoggerFactory.getLogger(LoadScoreWeightDBServiceImpl.class);

    private String driver;
    private String user;
    private String pass;
    private String sql;

    private String url;

    public String getDriver() {
        return driver;
    }

    public void setDriver(String driver) {
        this.driver = driver;
    }

    public String getUser() {
        return user;
    }

    public void setUser(String user) {
        this.user = user;
    }

    public String getPass() {
        return pass;
    }

    public void setPass(String pass) {
        this.pass = pass;
    }

    public String getUrl() {
        return url;
    }

    public void setUrl(String url) {
        this.url = url;
    }

    public String getSql() {
        return sql;
    }

    public void setSql(String sql) {
        this.sql = sql;
    }

    @Override
    public JavaPairRDD<String, ScoreWeight> getScoreWeightRdd(SparkSession session) {
        Properties properties = new Properties();
        properties.put("driver", this.driver);
        properties.put("user", this.user);
        properties.put("password", this.pass);

        String table = getTable();

        Dataset<Row> weigths = session.read().jdbc(
                this.url,
                table,
                properties);

        return weigths.toJavaRDD().mapPartitionsToPair(
                (PairFlatMapFunction<Iterator<Row>, String, ScoreWeight>) LoadScoreWeightDBServiceImpl::getScoreWeightPair);
    }
    public Map<String, ScoreWeight> getScoreWeight() {
//        Properties properties = new Properties();
//        properties.put("driver", this.driver);
//        properties.put("user", this.user);
//        properties.put("password", this.pass);
        Map<String, ScoreWeight> result = new HashMap<>();
        try {
            Class.forName(driver);
            Connection conn = DriverManager.getConnection(url, user, pass);
            Statement statement = conn.createStatement();
            ResultSet rs = statement.executeQuery(sql);

            while (rs.next()) {
                // 载重code
                String load_code = rs.getString(1);
                // 评分类型
                String type = rs.getString(2);
                // 一级分类
                String level1 = rs.getString(3);
                // 二级分类
                String level2 = rs.getString(4);
                // 指标code
                String modelCode = rs.getString(5);
                // 指标类型
                String modelType = rs.getString(6);
                // 指标
                String modelName = rs.getString(7);
                // 权重
                String weightStr = rs.getString(8);
                if(StringUtils.isEmpty(weightStr)) continue;
                BigDecimal weight = new BigDecimal(weightStr);
                // 公式
                String formula = rs.getString(9);
                // 参数数量
                int num = rs.getInt(10);
                // int num = StringUtils.isEmpty(numStr) ? 0 : Integer.valueOf(numStr);
                // 参数1
                String param1 = rs.getString(11);
                // 参数2
                String param2 = rs.getString(12);
                // 参数3
                String param3 = rs.getString(13);
                // 参数4
                String param4 = rs.getString(14);
                // 参数5
                String param5 = rs.getString(15);
                String param6 = rs.getString(16);

                ScoreWeight scoreWeight = new ScoreWeight();
                scoreWeight.setLoad_code(load_code);
                scoreWeight.setType(type);
                scoreWeight.setLevel1(level1);
                scoreWeight.setLevel2(level2);
                scoreWeight.setModelCode(modelCode);
                scoreWeight.setModelType(modelType);
                scoreWeight.setModelName(modelName);
                scoreWeight.setWeight(weight);
                scoreWeight.setFormula(formula);
                scoreWeight.setNum(num);
                scoreWeight.setParam1(param1);
                scoreWeight.setParam2(param2);
                scoreWeight.setParam3(param3);
                scoreWeight.setParam4(param4);
                scoreWeight.setParam5(param5);
                scoreWeight.setParam6(param6);
                String key = load_code + "_" + level2 + "_" + modelType;
                result.put(key, scoreWeight);
            }
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        } catch (SQLException e) {
            logger.error("SQLException", e);
        }
        return result;
    }
    private static Iterator<Tuple2<String, ScoreWeight>> getScoreWeightPair(Iterator<Row> iterator) {
        List<Tuple2<String, ScoreWeight>> vehicleInfos = new ArrayList<>();
        while (iterator.hasNext()) {
            Row next = iterator.next();
            // 载重code
            String load_code = next.getString(0);
            // 评分类型
            String type = next.getString(1);
            // 一级分类
            String level1 = next.getString(2);
            // 二级分类
            String level2 = next.getString(3);
            // 指标code
            String modelCode = next.getString(4);
            // 指标类型
            String modelType = next.getString(5);
            // 指标
            String modelName = next.getString(6);
            // 权重
            BigDecimal weight = next.getDecimal(7);
//            double weight = StringUtils.isEmpty(weightStr) ? 0 : Double.valueOf(weightStr);
            // 公式
            String formula = next.getString(8);
            // 参数数量
            int num = next.getInt(9);
            // int num = StringUtils.isEmpty(numStr) ? 0 : Integer.valueOf(numStr);
            // 参数1
            String param1 = next.getString(10);
            // 参数2
            String param2 = next.getString(11);
            // 参数3
            String param3 = next.getString(12);
            // 参数4
            String param4 = next.getString(13);
            // 参数5
            String param5 = next.getString(14);
            String param6 = next.getString(15);

            ScoreWeight scoreWeight = new ScoreWeight();
            scoreWeight.setLoad_code(load_code);
            scoreWeight.setType(type);
            scoreWeight.setLevel1(level1);
            scoreWeight.setLevel2(level2);
            scoreWeight.setModelCode(modelCode);
            scoreWeight.setModelType(modelType);
            scoreWeight.setModelName(modelName);
            scoreWeight.setWeight(weight);
            scoreWeight.setFormula(formula);
            scoreWeight.setNum(num);
            scoreWeight.setParam1(param1);
            scoreWeight.setParam2(param2);
            scoreWeight.setParam3(param3);
            scoreWeight.setParam4(param4);
            scoreWeight.setParam5(param5);
            scoreWeight.setParam6(param6);
            String key = load_code + "_" + level2 + "_" + modelType;
            vehicleInfos.add(new Tuple2<>(key, scoreWeight));
        }
        return vehicleInfos.iterator();
    }

    private String getTable() {
        return "(SELECT \n" +
                "    sw.loadCode,\n" +
                "    sw.type,\n" +
                "    sw.level1,\n" +
                "    sw.level2,\n" +
                "    sw.modelCode,\n" +
                "    sw.modelType,\n" +
                "    sw.modelName,\n" +
                "    sw.weight,\n" +
                "    sf.formula,\n" +
                "    sf.num,\n" +
                "    sf.param1,\n" +
                "    sf.param2,\n" +
                "    sf.param3,\n" +
                "    sf.param4,\n" +
                "    sf.param5,\n" +
                "    sf.param6\n" +
                "FROM\n" +
                "    score_weight sw\n" +
                "  JOIN\n" +
                "    score_formula sf\n" +
                "  ON\n" +
                "    sw.modelType = sf.modelType AND sw.enable = 1 AND sf.enable = 1\n" +
                ") t";
    }
}
